package com.mikelouis.test.scala.impatient.chapter20_actor

/**
 * Created by yuazhu on 5/11/2015.
 */
object Chapter20 extends App {
  // 20.1

  import scala.actors.Actor

  case class InitMessage(size: Int, initSplit: Int)

  // message used for array generation
  case class GenerateMessage(size: Int)

  case class GenerateDone(array: Array[Int])

  // message used for average calculation
  case class AverageMessage(arrayToCal: Array[Int])

  case class AverageCalDone(size: Int, avg: Double)

  class GenerateActor extends Actor {
    override def act(): Unit = {
      while (true) {
        receive {
          case GenerateMessage(size) =>
            val generatedArray = new Array[Int](size)
            for (i <- 0 until size) generatedArray(i) = i
            sender ! GenerateDone(generatedArray)
        }
      }
    }
  }

  class AverageCalActor extends Actor {
    override def act(): Unit = {
      while (true) {
        receive {
          case AverageMessage(arrayToCal) =>
            // need to convert to long to avoid out of range of Int
            val sumArr: Long = arrayToCal.map(_.toLong).sum
            sender ! AverageCalDone(arrayToCal.length, sumArr / arrayToCal.length)
        }
      }
    }
  }

  class GenerateSummarizer(completion: Actor, generatorActor: Seq[Actor], totalSize: Int, split: Int) extends Actor {
    var count = 0
    var arrayGenerated: Array[Int] = Array[Int]()

    def updateCount(n: Int): Unit = {
      count += n
      if (count == 0) completion ! GenerateDone(arrayGenerated)
    }

    override def act(): Unit = {
      while (true) {
        receive {
          case InitMessage(size, initSplit) =>
            // initialize count
            updateCount(initSplit)
            // send message to generate actor for generating error
            for (a <- generatorActor) a ! GenerateMessage(totalSize / split)
          case GenerateDone(generatedArray) =>
            arrayGenerated ++= generatedArray
            updateCount(-1)
        }
      }
    }
  }

  class CalculateAverageSummarizer(completion: Actor, calAvgActors: Seq[Actor], arrayToCal: Array[Int], split: Int) extends Actor {
    var count: Int = 0
    var size: Int = 0
    var avg: Double = 0

    def updateCount(n: Int): Unit = {
      count += n
      if (count == 0) {
        completion ! AverageCalDone(size, avg)
      }
    }

    override def act(): Unit = {
      while (true) {
        receive {
          case InitMessage(avgSize, avgSplit) =>
            updateCount(avgSplit)
            calAvgActors.zip(arrayToCal.grouped(avgSize / avgSplit).toList).map {
              case (actor, arrToCal) => actor ! AverageMessage(arrToCal)
            }
          case AverageCalDone(subSize, subAvg) =>
            avg = (avg * size + subSize * subAvg) / (size + subSize)
            size += subSize
            println(s"subList average cal -> $subSize, $subAvg")
            println(s"List average cal -> $size, $avg")
            updateCount(-1)
        }
      }
    }
  }

  // How many random number are going to be generated
  val n = 2 * 1024 * 1024
  val intSplit = 8

  // start generate actors
  val generateActors = for (i <- 1 to intSplit) yield new GenerateActor
  for (a <- generateActors) a.start()

  // start average calculators
  val avgActors = for (i <- 1 to intSplit) yield new AverageCalActor
  avgActors.foreach(_.start())

  import scala.actors.Actor._

  actor {
    // start the generate summarizer
    val genSum = new GenerateSummarizer(self, generateActors, n, intSplit)
    genSum.start()

    genSum ! InitMessage(n, intSplit)

    //      receive {
    //        case GenerateDone(arr) =>
    //          println(arr.mkString("[", ",", "]"))
    //          System.exit(0)
    //      }

    while (true) {
      receive {
        case GenerateDone(arr) =>
          println(s"Array Generated, length=${arr.length}")
          val avgSum = new CalculateAverageSummarizer(self, avgActors, arr, intSplit)
          avgSum.start()
          avgSum ! InitMessage(n, intSplit)
        case AverageCalDone(size, avg) =>
          println(size, avg)
          System.exit(0)
      }
    }
  }
}
